package com.shujia.spark.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SparkSession}

object Demo11Shebao {
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("shebao")
      .config("spark.sql.shuffle.partitions", 1) //指定spark sql shuffle之后rdd的分区数
      .getOrCreate()

    //读取数据
    val shebaoDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("id STRING , item STRING , sdate STRING")
      .load("data/shebao.txt")
    import spark.implicits._
    import org.apache.spark.sql.functions._

    shebaoDF
      //取出月份
      .withColumn("s_month", date_format($"sdate", "yyyy-MM-dd"))
      //取出这个人上一个月所在的公司
      .withColumn("last_item", lag($"item", 1) over Window.partitionBy($"id").orderBy($"s_month"))
      //在还工作的时候加上一个标记， 换工作时加上1，没有换工作加上0
      .withColumn("flag", when($"item" === $"last_item", 0).otherwise(1))
      //对标记累加求和，得到类别
      .withColumn("clazz", sum($"flag") over Window.partitionBy($"id").orderBy($"s_month"))
      //按照id ,公司，clazz分组
      .groupBy($"id", $"item", $"clazz")
      //获取入职月份和离职月份
      .agg(min($"s_month") as "start_month", max($"s_month") as "end_month")
      .show()

  }

}
